package com.study.flink.datastream

import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, StreamExecutionEnvironment}

/**
  * Connect操作
  *
  * @author stephen
  * @date 2019-07-19 16:50
  */
object FlinkConnectDemo {

  def main(args: Array[String]): Unit = {
    // 1、执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2、指定数据源
    import org.apache.flink.api.scala._
    val dataStream1: DataStream[String] = env.fromCollection(List("a", "b", "c", "d"))
    val dataStream2: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5, 6)

    // 3 处理
    val connectStream: ConnectedStreams[String, Int] = dataStream1.connect(dataStream2)
    import org.apache.flink.api.scala._
    // map函数中的第一个函数作用于 ConnectedStreams 的第一个 DataStream;第二个函数作用于第二个 DataStream
    connectStream.map(e => println("1->" + e), e => println("2->" + e))
    //connectStream.flatMap()

    // 4 启动任务
    env.execute("Flink Connect Demo")
  }
}
